package com.somg.kafka_study.demo_01;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class ProductorDemo {

	
	public static void main(String[] args) {
		Properties props = new Properties();
		props.put("zk.connect", "localhost:2181");
		props.put("metadata.broker.list", "localhost:9092");
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		props.put("key.serializer.class", "kafka.serializer.StringEncoder");
		
//		props.put("partitioner.class", "")  可选配置， 如果不配置 则默认的partitioner
		// 触发acknoewledgement 机制， 否则  and forget 可能引起数据的丢失  值为0 ， 1， -1 
		props.put("request.required.acks", "1");
//		KafkaProducer<String,String> producer  = new KafkaProducer<String,String>(props);
//		ProducerRecord<String, String> recoder = new ProducerRecord<String, String>("test", "hello word");
//		producer.send(recoder);
		
		ProducerConfig config =new  ProducerConfig(props);
		
		Producer<String, String> producer = new Producer<String, String>(config);
		int i = 0;
		while(true)
		{
			
			producer.send(new KeyedMessage<String, String>("test", "hello word"+i));
			i++;
			
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
		
		
//		producer.close();
	
	}
	
}
